package cn.langpy.dblistener.core;

import cn.langpy.dblistener.core.model.DbType;
import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

public class ListenEngine implements Runnable {
    private static Logger log = Logger.getLogger(ListenEngine.class.toString());

    private Executor executor;
    private Integer parseQueueSize;
    private Integer maxBatchSize;
    private Integer parseThreadSize;
    private Integer processThreadSize;
    private DbType type;
    private String driver;
    private String host;
    private Integer port;
    private String username;
    private String password;
    private List<String> databases;
    private List<String> tables;
    private List<String> tablesExclude;
    private Boolean initialize;
    private Boolean multiply;
    private List<EventListener> listeners = new ArrayList<>();

    final Map<DbType, String> connectorMap = new HashMap<DbType, String>() {
        {
            put(DbType.Mysql, "io.debezium.connector.mysql.MySqlConnector");
            put(DbType.Postgres, "io.debezium.connector.postgresql.PostgresConnector");
        }
    };

    public ListenEngine(Builder builder) {
        this.parseQueueSize = builder.parseQueueSize;
        this.maxBatchSize = builder.maxBatchSize;
        this.parseThreadSize = builder.parseThreadSize;
        this.processThreadSize = builder.processThreadSize;
        this.type = builder.type;
        this.host = builder.host;
        this.port = builder.port;
        this.username = builder.username;
        this.password = builder.password;
        this.databases = builder.databases;
        this.tables = builder.tables;
        this.tablesExclude = builder.tablesExclude;
        this.listeners.addAll(builder.listeners);
        if (builder.executor==null) {
            this.executor = Executors.newFixedThreadPool(processThreadSize);
        }
        this.initialize = builder.initialize;
        this.driver = builder.driver;
        this.multiply = builder.multiply;
    }

    public static Builder builder() {
        return new Builder();
    }

    public static class Builder {
        private Integer parseQueueSize = 8192;
        private Integer maxBatchSize = 2048;
        private Integer parseThreadSize = 2;
        private Integer processThreadSize = 2;
        private DbType type;
        private String driver;
        private String host;
        private Integer port;
        private String username;
        private String password;
        private List<String> databases;
        private List<String> tables;
        private List<String> tablesExclude;
        private List<EventListener> listeners = new ArrayList<>();
        private Executor executor;
        private Boolean initialize = true;
        private Boolean multiply = false;

        public Builder multiply(Boolean multiply) {
            this.multiply = multiply;
            return this;
        }
        public Builder driver(String driver) {
            this.driver = driver;
            return this;
        }

        public Builder initialize(Boolean initialize) {
            this.initialize = initialize;
            return this;
        }

        public Builder executor(Executor executor) {
            this.executor = executor;
            return this;
        }
        public Builder tablesExclude(List<String> tablesExclude) {
            this.tablesExclude = tablesExclude;
            return this;
        }
        public Builder listener(EventListener listener){
            this.listeners.add(listener);
            return this;
        }
        public Builder listeners(List<? extends EventListener> listeners){
            this.listeners.addAll(listeners);
            return this;
        }
        public Builder parseQueueSize(Integer parseQueueSize) {
            this.parseQueueSize = parseQueueSize;
            return this;
        }

        public Builder maxBatchSize(Integer maxBatchSize) {
            this.maxBatchSize = maxBatchSize;
            return this;
        }

        public Builder parseThreadSize(Integer parseThreadSize) {
            this.parseThreadSize = parseThreadSize;
            return this;
        }

        public Builder processThreadSize(Integer processThreadSize) {
            this.processThreadSize = processThreadSize;
            return this;
        }

        public Builder type(DbType type) {
            this.type = type;
            return this;
        }

        public Builder host(String host) {
            this.host = host;
            return this;
        }

        public Builder port(Integer port) {
            this.port = port;
            return this;
        }

        public Builder username(String username) {
            this.username = username;
            return this;
        }

        public Builder password(String password) {
            this.password = password;
            return this;
        }

        public Builder databases(List<String> databases) {
            this.databases = databases;
            return this;
        }

        public Builder tables(List<String> tables) {
            this.tables = tables;
            return this;
        }

        public ListenEngine build() {
            return new ListenEngine(this);
        }
    }

    public DebeziumEngine<ChangeEvent<String, String>> create(List<String> databases) {
        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
                .using(createConfig(databases))
                .notifying(new ChangeConsumer(this.type,this.listeners,this.tablesExclude,this.executor))
                .using((success, msg, error) -> {
                    log.severe("【DBListener】=>监听错误：" + error.getMessage());
                }).build();
        closeOnExit(engine);
        return engine;
    }

    private void closeOnExit(DebeziumEngine<ChangeEvent<String, String>> engine) {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                engine.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }));
    }

    @Override
    public void run() {
        if (this.multiply) {
            for (String database : this.databases) {
                DebeziumEngine<ChangeEvent<String, String>> changeEngine = create(Arrays.asList(database));
                changeEngine.run();
            }
        }else {
            DebeziumEngine<ChangeEvent<String, String>> changeEngine = create(this.databases);
            changeEngine.run();
        }
    }

    public void start() {
        start(false);
    }

    public void start(boolean blocked) {
        final ExecutorService executor = Executors.newFixedThreadPool(databases.size());
        if (multiply) {
            for (String database : this.databases) {
                DebeziumEngine<ChangeEvent<String, String>> changeEngine = create(Arrays.asList(database));
                executor.execute(changeEngine);
            }
        }else {
            DebeziumEngine<ChangeEvent<String, String>> changeEngine = create(this.databases);
            executor.execute(changeEngine);
        }
        if (blocked) {
            try {
                executor.awaitTermination(Integer.MAX_VALUE, TimeUnit.DAYS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public Properties createConfig(List<String> databases) {
        final Properties props = new Properties();
        props.setProperty("snapshot.mode", this.initialize?"initial":"no_data");
        props.setProperty("record.processing.threads", this.parseThreadSize + "");
        props.setProperty("snapshot.max.threads", this.parseThreadSize + "");
        /*读取记录的模式：{UNORDERED，ORDERED}*/
        props.setProperty("record.processing.order", "ORDERED");
        props.setProperty("max.queue.size", this.parseQueueSize + "");
        props.setProperty("max.batch.size", this.maxBatchSize + "");
        props.setProperty("snapshot.fetch.size", "512");
        props.setProperty("min.row.count.to.stream.results", "0");
        props.setProperty("snapshot.locking.mode", "none");
        props.setProperty("name", "huoyo-" + new Random().nextInt(99));
        props.setProperty("metrics.group-prefix", "huoyo-metric-" + new Random().nextInt(99));
        props.setProperty("connector.class", connectorMap.get(this.type));
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.storage.file.filename", "/path/to/storage/offsets.dat");
        props.setProperty("offset.flush.interval.ms", "0");
        props.setProperty("min.row.count.to.stream.results", "200");
        props.setProperty("snapshot.tables.order.by.row.count", "ascending");
        props.setProperty("table.ignore.builtin", "true");
        props.setProperty("database.hostname", this.host);
        props.setProperty("database.port", this.port + "");
        props.setProperty("database.user", this.username);
        props.setProperty("database.password", this.password);
        props.setProperty("database.jdbc.driver", this.driver==null?"com.mysql.cj.jdbc.Driver":this.driver);
        props.setProperty("database.server.id", "30438" + new Random().nextInt(99));
        props.setProperty("database.ssl.mode", "disabled");
        props.setProperty("database.include.list", String.join(",", databases));
        props.setProperty("table.include.list", String.join(",", this.tables));
        props.setProperty("include.schema.changes", "true");
        props.setProperty("include.schema.comments", "true");
        props.setProperty("bigint.unsigned.handling.mode", "long");
        props.setProperty("decimal.handling.mode", "double");
        props.setProperty("time.precision.mode", "adaptive_time_microseconds");
        props.setProperty("binary.handling.mode", "bytes");
        props.setProperty("topic.prefix", "huoyo-connector");
        props.setProperty("schemas.enable", "true");
        props.setProperty("schema.history.internal.store.only.captured.tables.ddl", "true");
        props.setProperty("schema.history.internal.store.only.captured.databases.ddl", "true");
        props.setProperty("schema.history.internal", "io.debezium.storage.file.history.FileSchemaHistory");
        props.setProperty("schema.history.internal.file.filename", "/path/to/storage/schemahistory.dat");
        return props;
    }
}
